package com.tc.vms.iface;

import com.tc.vms.SLA;
import com.tc.vms.config.VmsClientContext;
import com.tc.vms.config.VmsRouterManager;
import com.tc.vms.connection.VmsConnectionManager;
import com.tc.vms.consumer.DlqQueueMetaData;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.message.MessageIdGenerator;
import com.tc.vms.metric.VmsMetricsGroup;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.VmsThreadFactory;
import com.tc.vms.utils.ZkUtils;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public abstract class VmsConnectionWrap extends VmsMetricsGroup
        implements IZkStateListener, IZkDataListener, VmsService {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    protected final VmsConnectionManager vmsConnectionManager;
    protected final SLA serviceSLA;
    protected final VmsQos vmsQos;
    protected final Map<?, ?> userProperties;
    protected final MessageIdGenerator msgGenerator;
    private final ScheduledExecutorService tpsLimitTimer = Executors
            .newSingleThreadScheduledExecutor(new VmsThreadFactory("VmsConnectionWrap-TpsLimitTimer"));

    protected volatile VmsMetaData vmsMetaData;
    protected volatile boolean isOpen = false;
    protected volatile boolean isSuspend = false;
    protected volatile int tpsLimit = 0;
    protected volatile int sleepIntervalWhenTpsLimit = 1;
    protected String zkEphemeralPath;

    public String getZkEphemeralPath() {
        return zkEphemeralPath;
    }

    public void setZkEphemeralPath(String zkEphemeralPath) {
        this.zkEphemeralPath = zkEphemeralPath;
    }

    public VmsConnectionWrap(VmsConnectionManager vmsConnectionManager, VmsMetaData vmsMetaData, SLA serviceSLA,
                             VmsQos vmsQos, Properties props) {
        this.vmsConnectionManager = vmsConnectionManager;
        this.vmsMetaData = vmsMetaData;
        this.isSuspend = vmsMetaData.isSuspend;
        this.serviceSLA = serviceSLA;
        this.vmsQos = vmsQos;
        this.msgGenerator = new MessageIdGenerator();
        this.userProperties = (props == null) ?
                Collections.unmodifiableMap(new HashMap<String, String>(0)) :
                Collections.unmodifiableMap(props);
        if (this.isSuspend) {
            LOGGER.info("ControlCommand [isSuspend = true], VMS instance [{}] would suspend.",
                    vmsMetaData.getClientInstName());
        }

        if (vmsMetaData.isLoggerOpen) {
            LOGGER.info("ControlCommand [isLoggerOpen = true], VMS instance [{}] logger would change to debug model.",
                    vmsMetaData.getClientInstName());
            VmsClientLogger.setLoggerLevel(vmsMetaData.isLoggerOpen);
        }

        validTpsLimitAndSet(vmsMetaData.getTpsLimit());

        tpsLimitTimer.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    adjustSleepIntervalWhenTpsLimit();
                } catch (Exception e) {
                    LOGGER.warn("adjustSleepIntervalWhenTpsLimit exception.", e);
                }
            }
        }, 15 * 1000, 30 * 1000, TimeUnit.MILLISECONDS);
    }

    private void adjustSleepIntervalWhenTpsLimit() {
        if (this.tpsLimit > 0) {
            double realTps = getRealTpsInOneMinute();
            LOGGER.debug("VmsProducerProxy adjustSleepIntervalWhenTpsLimit realTps=[{}], tpsLimit=[{}].", realTps,
                    this.tpsLimit);
            if (realTps > this.tpsLimit * 1.5d) {
                this.sleepIntervalWhenTpsLimit = this.sleepIntervalWhenTpsLimit < Utils.VMS_MAX_TPS_LIMIT ?
                        this.sleepIntervalWhenTpsLimit << 1 :
                        Utils.VMS_MAX_TPS_LIMIT;
                LOGGER.debug("VmsProducerProxy adjustSleepIntervalWhenTpsLimit tpsLimit=[{}], sleepInterval=[{}] ms.",
                        tpsLimit, sleepIntervalWhenTpsLimit);
            } else if ((realTps * 1.5d) < this.tpsLimit && this.sleepIntervalWhenTpsLimit > (1 << 1)) {
                this.sleepIntervalWhenTpsLimit = this.sleepIntervalWhenTpsLimit >> 1;
                LOGGER.debug("VmsProducerProxy adjustSleepIntervalWhenTpsLimit tpsLimit=[{}], sleepInterval=[{}] ms.",
                        tpsLimit, sleepIntervalWhenTpsLimit);
            }
        } else {
            this.sleepIntervalWhenTpsLimit = 1;
        }
    }

    protected abstract double getRealTpsInOneMinute();

    private void validTpsLimitAndSet(int tpsLimit) {
        this.tpsLimit = tpsLimit;
        if (this.tpsLimit > 0) {
            LOGGER.info("ControlCommand [vms.tps.limit = {}], VMS would change to limit tps.", this.tpsLimit);
            if (this.tpsLimit > Utils.VMS_MAX_TPS_LIMIT) {
                LOGGER.info("the value of vms.tps.limit is too large, would be modified to [{}].",
                        Utils.VMS_MAX_TPS_LIMIT);
                this.tpsLimit = Utils.VMS_MAX_TPS_LIMIT;
            }
        } else if (this.tpsLimit < 0) {
            LOGGER.info("ControlCommand [vms.tps.limit = {}] is invalid and reset to 0, VMS couldn't limit tps.",
                    this.tpsLimit);
            this.tpsLimit = 0;
        }
    }

    public final VmsClientContext getVmsClientContext() {
        return vmsConnectionManager.getVmsRouterManager().getVmsClientContext();
    }

    public void setVmsMetaData(VmsMetaData vmsMetaData) {
        this.vmsMetaData = vmsMetaData;
    }

    public Map<?, ?> getUserProps() {
        return this.userProperties;
    }

    public boolean isSuspend() {
        return isSuspend;
    }

    /**
     * 注册producer 和consumer 临时节点，监听channel和queue以及cluster变更信息
     */
    public synchronized void registerZkNode() {
        try {
            vmsConnectionManager.getVmsRouterManager().registerWatcherHandler(this);
            vmsConnectionManager.getVmsRouterManager().registerInstEphemeralPath(this);
        } catch (Exception ex) {
            LOGGER.warn("registerZkNode error.", ex);
        }
    }

    /**
     * 反注册producer 和consumer临时节点，取消监听channel和queue以及cluster变更信息
     */
    public synchronized void unregisterZkNode() {
        try {
            vmsConnectionManager.getVmsRouterManager().unregisterWatcherHandler(this);
            vmsConnectionManager.getVmsRouterManager().unregisterInstEphemeralPath(this);
        } catch (Exception ex) {
            LOGGER.warn("unregisterZkNode error.", ex);
        }
    }

    public <T extends VmsMetaData> T getVmsMetaData() {
        return (T) vmsMetaData;
    }

    public <T extends VmsQos> T qos() {
        return (T) vmsQos;
    }

    public <T extends SLA> T sla() {
        return (T) serviceSLA;
    }

    protected void chkResetOption() throws VmsClientException {
        if (isOpen) {
            throw new VmsClientException("Modifying the options must do before running",
                    VmsClientErrorCode.VMS_SYS_ERR_MODIFY_OPTION);
        }
    }

    public final boolean isOpen() {
        return isOpen;
    }

    public abstract void zkNotifyReconnectServer(VmsMetaData newMetaData, VmsMetaData origMetaData);

    /**
     * handleDataChange process channel & queue node changes event and cluster node changes
     *
     * @param dataPath
     * @param data     content stored in znode
     * @throws Exception
     */
    @Override
    public synchronized void handleDataChange(String dataPath, Object data) throws Exception {
        VmsRouterManager routerManager = vmsConnectionManager.getVmsRouterManager();
        VmsMetaData newVmsMetaData = null;
        if (vmsMetaData.isChannelMetaData()) {
            newVmsMetaData = routerManager
                    .readChannelMetaData(vmsMetaData.getZnodeName(), vmsMetaData.getOrigClientId());
        } else { // queue event
            if (vmsMetaData instanceof DlqQueueMetaData) {
                newVmsMetaData = routerManager
                        .readDLQQueueMetaData(vmsMetaData.getZnodeName(), vmsMetaData.getOrigClientId());
            } else {
                newVmsMetaData = routerManager
                        .readQueueMetaData(vmsMetaData.getZnodeName(), vmsMetaData.getOrigClientId());
            }
        }

        //judge current instant would be ignored.
        try {
            /*if (newVmsMetaData.isGrayMatched()) {
            LOGGER.info("Current instance would process zk notify message, for it's in gay-publish list. ip=[{}]",
					MessageConst.HOSTNAME);*/
            handleControlCommand(newVmsMetaData, vmsMetaData);
            handleCommonCommand(newVmsMetaData, vmsMetaData);
            /*} else {
                LOGGER.info("Current instance will ignore zk notify message, for it's not in gay-publish list. ip=[{}]",
						MessageConst.HOSTNAME);
			}*/
        } catch (Exception ex) {
            LOGGER.warn(
                    "handleDataChange orig = {} has been changed, result new = {}, because VmsClientFilter pattern error.",
                    vmsMetaData, newVmsMetaData, ex);
        }
    }

    private synchronized void handleControlCommand(VmsMetaData newMetaData, VmsMetaData oldMetaData) {
        Properties newControlProps = newMetaData.getControlProps();
        if (newControlProps.isEmpty()) {
            LOGGER.info("ControlCommand has been changed, values is empty, restore properties.");
            this.isSuspend = false;
            this.tpsLimit = 0;
            VmsClientLogger.setLoggerLevel(false);
            return;
        }
        LOGGER.info("ControlCommand has been changed, new = {}.", newControlProps);

        if (newControlProps.containsKey(ZkUtils.vmsSuspendSwitcher)) {
            this.isSuspend = ZkUtils.getVmsIsSuspend(newControlProps);
        } else {
            this.isSuspend = false;
        }

        if (newControlProps.containsKey(ZkUtils.vmsTpsLimit)) {
            validTpsLimitAndSet(ZkUtils.getVmsTpsLimit(newControlProps));
        } else {
            this.tpsLimit = 0;
        }

        if (newControlProps.containsKey(ZkUtils.vmsLoggerSwitcher)) {
            boolean isDebugLevel = ZkUtils.getVmsLoggerSwitcher(newControlProps);
            VmsClientLogger.setLoggerLevel(isDebugLevel);
        } else {
            VmsClientLogger.setLoggerLevel(false);
        }
    }

    private synchronized void handleCommonCommand(VmsMetaData newMetaData, VmsMetaData oldMetaData) {
        Properties newCommonProps = newMetaData.getCommonProps();
        Properties oldCommonProps = oldMetaData.getCommonProps();
        if (oldCommonProps.equals(newCommonProps))
            return;
        LOGGER.info("CommonCommand orig = {} has been changed, result new = {}.", oldCommonProps, newCommonProps);
        zkNotifyReconnectServer(newMetaData, oldMetaData);
    }

    @Override
    public synchronized void handleDataDeleted(String dataPath) throws Exception {
        //invoke vmsConnectionManager to shutdown wrapper and remove vmsconnection
        vmsConnectionManager.removeConnection(this);
    }

    protected void chkServiceOk() throws VmsClientException {
        if (!isOpen) {
            throw new VmsClientException("Service isn't running.", VmsClientErrorCode.VMS_SYS_ERR_NOT_RUNNING);
        }

        if (isSuspend) {
            throw new VmsClientException("Service is suspending by vmscenter.",
                    VmsClientErrorCode.VMS_SYS_RUNTIME_ERR_SUSPEND_SERVICE);
        }

        if (tpsLimit > 0) {
            Utils.sleep(sleepIntervalWhenTpsLimit);
        }

    }

    private void suspendFunc() throws VmsClientException {
        if (isOpen && isSuspend) {
			/*while (true) {
				if (isOpen && isSuspend) {
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) {
                    }
                } else {
                    break;
                }
            }*/
        }
    }

    @Override
    public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {

    }

    @Override
    public void handleNewSession() throws Exception {
        //when reconnected, re-create the ephemeral nodes
        vmsConnectionManager.getVmsRouterManager().registerInstEphemeralPath(this);
    }

    @Override
    public void shutdown() {
        this.tpsLimitTimer.shutdown();
        unregisterZkNode();
    }
}
